package cn._51doit;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 标准键盘布局
 * ┌───┐   ┌───┬───┬───┬───┐ ┌───┬───┬───┬───┐ ┌───┬───┬───┬───┐ ┌───┬───┬───┐
 * │Esc│   │ F1│ F2│ F3│ F4│ │ F5│ F6│ F7│ F8│ │ F9│F10│F11│F12│ │P/S│S L│P/B│  ┌┐    ┌┐    ┌┐
 * └───┘   └───┴───┴───┴───┘ └───┴───┴───┴───┘ └───┴───┴───┴───┘ └───┴───┴───┘  └┘    └┘    └┘
 * ┌───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───────┐ ┌───┬───┬───┐ ┌───┬───┬───┬───┐
 * │~ `│! 1│@ 2│# 3│$ 4│% 5│^ 6│& 7│* 8│( 9│) 0│_ -│+ =│ BacSp │ │Ins│Hom│PUp│ │N L│ / │ * │ - │
 * ├───┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─────┤ ├───┼───┼───┤ ├───┼───┼───┼───┤
 * │ Tab │ Q │ W │ E │ R │ T │ Y │ U │ I │ O │ P │{ [│} ]│ | \ │ │Del│End│PDn│ │ 7 │ 8 │ 9 │   │
 * ├─────┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴─────┤ └───┴───┴───┘ ├───┼───┼───┤ + │
 * │ Caps │ A │ S │ D │ F │ G │ H │ J │ K │ L │: ;│" '│ Enter  │               │ 4 │ 5 │ 6 │   │
 * ├──────┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴────────┤     ┌───┐     ├───┼───┼───┼───┤
 * │ Shift  │ Z │ X │ C │ V │ B │ N │ M │< ,│> .│? /│  Shift   │     │ ↑ │     │ 1 │ 2 │ 3 │   │
 * ├─────┬──┴─┬─┴──┬┴───┴───┴───┴───┴───┴──┬┴───┼───┴┬────┬────┤ ┌───┼───┼───┐ ├───┴───┼───┤ E││
 * │ Ctrl│    │Alt │         Space         │ Alt│    │    │Ctrl│ │ ← │ ↓ │ → │ │   0   │ . │←─┘│
 * └─────┴────┴────┴───────────────────────┴────┴────┴────┴────┘ └───┴───┴───┘ └───────┴───┴───┘
 *
 * @create: 2021-10-13 19:39
 * @author: 今晚打脑斧
 * @program: ProducerAckDemo
 * @Description:
 *   保证生产者写入数据是安全的
 *   设置acks为 1 或 -1/all
 *   同时还要设置retries大于0
 *       0(渣男)       :生产者向leader分区发送数据,但是不等待任何响应,就可以继续发送数据.
 *       -1/all(暖男)  :不但要leader将数据落盘,还要等待所有的follower分区数据罗盘,然后响应leader分区,最后leader分区响应生产者,生产者才会继续写数据
 *       1            :就是生产者将数据发送给leader分区,leader分区将数据罗盘即响应producer
 *   ACKS如果等于-1/all最安全,但是效率低
 *   ACKS如果等于1相对安全, 效率相对较高
 *   ACKS如果等于0最不安全,但是效率最高
 **/
public class JavaProducerAckDemo {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // 连接kafka节点
        properties.setProperty("bootstrap.servers", "doit01:9092,doit02:9092,doit03:9092");
        //指定key序列化方式
        properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //指定value序列化方式
        properties.setProperty("value.serializer", StringSerializer.class.getName()); // 两种写法都行


        //设置ACK
        properties.setProperty("acks","1");
        //设置retries  失败后尝试重新写的次数
        properties.setProperty("retries",Integer.MAX_VALUE+"");


        //创建一个生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //封装数据
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("wordcount", null, "今晚打脑斧");
        producer.send(producerRecord);
        producer.close();
    }
}
